package com.xiaoyun.service.impl;

import com.xiaoyun.config.HBaseClient;
import com.xiaoyun.constant.Constants;
import com.xiaoyun.dao.MerchantsDao;
import com.xiaoyun.entity.Merchants;
import com.xiaoyun.mapper.PassTemplateRowMapper;
import com.xiaoyun.service.IInventoryService;
import com.xiaoyun.service.IUserPassService;
import com.xiaoyun.utils.RowKeyGenUtil;
import com.xiaoyun.vo.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.LongComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 获取库存信息 只返回用户没有领取的
 */
@Service
@Slf4j
public class InventoryServiceImpl implements IInventoryService {

    private final HBaseClient hbaseTemplate;

    private final MerchantsDao merchantsDao;

    private final IUserPassService userPassService;

    @Autowired
    public InventoryServiceImpl(HBaseClient hbaseTemplate, MerchantsDao merchantsDao, IUserPassService userPassService) {
        this.hbaseTemplate = hbaseTemplate;
        this.merchantsDao = merchantsDao;
        this.userPassService = userPassService;
    }

    @Override
    public Response getInventoryInfo(Long useId) throws Exception {
        Response allPassInfo = userPassService.getUserAllPassInfo(useId);
        List<PassInfo> passInfos = (List<PassInfo>) allPassInfo.getData();
        List<PassTemplate> excludeObject = passInfos.stream().map(PassInfo::getPassTemplate).collect(Collectors.toList());
        List<String> excludeIds = new ArrayList<>();
        excludeObject.forEach(e -> excludeIds.add(
                RowKeyGenUtil.genPassTemplateRowKey(e)
        ));
        return new Response(new InventoryResponse(useId,
                buildPassTemplateInfo(getAvailablePassTemplate(excludeIds))));
    }

    /**
     * 获取系统中可用的优惠券
     *
     * @param excludeIds
     * @return
     */
    private List<PassTemplate> getAvailablePassTemplate(List<String> excludeIds) {
        FilterList filteredList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        filteredList.addFilter(
                new SingleColumnValueFilter(
                        Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                        Bytes.toBytes(Constants.PassTemplateTable.LIMIT),
                        CompareOperator.GREATER,
                        new LongComparator(0L)

                )
        );
        filteredList.addFilter(
                new SingleColumnValueFilter(
                        Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                        Bytes.toBytes(Constants.PassTemplateTable.LIMIT),
                        CompareOperator.EQUAL,
                        Bytes.toBytes(-1)

                )
        );
        Scan scan = new Scan();
        scan.setFilter(filteredList);

        List<PassTemplate> validTemplates = hbaseTemplate.find(Constants.PassTemplateTable.TABLE_NAME, scan, new PassTemplateRowMapper());
        List<PassTemplate> availablePassTemplates = new ArrayList<>();
        Date cur = new Date();
        for (PassTemplate validTemplate : validTemplates) {
            if (excludeIds.contains(RowKeyGenUtil.genPassTemplateRowKey(validTemplate))) {
                continue;
            }
            if (cur.getTime() >= validTemplate.getStart().getTime() && cur.getTime() <= validTemplate.getEnd().getTime()) {
                availablePassTemplates.add(validTemplate);
            }
        }
        return availablePassTemplates;
    }

    /**
     * 构造优惠券信息
     *
     * @param passTemplates
     * @return
     */
    private List<PassTemplateInfo> buildPassTemplateInfo(List<PassTemplate> passTemplates) {
        Map<Integer, Merchants> merchantsMap = new HashMap<>();
        List<Integer> merchantsIds = passTemplates.stream()
                .map(PassTemplate::getId).collect(Collectors.toList());
        List<Merchants> merchants = merchantsDao.findByIdIn(merchantsIds);
        merchants.forEach(m -> merchantsMap.put(m.getId(), m));
        List<PassTemplateInfo> result = new ArrayList<>(passTemplates.size());

        for (PassTemplate passTemplate : passTemplates) {
            Merchants mc = merchantsMap.getOrDefault(passTemplate.getId(), null);
            if (null == mc) {
                log.error("Merchants Error :{}", passTemplate.getId());
                continue;
            }
            result.add(new PassTemplateInfo(passTemplate, mc));
        }
        return result;

    }
}
